package com.apache.flink.basic

import com.apache.flink.bean.Domain.Access
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.api.scala._
import org.apache.flink.table.api.Table

object DataStreamTableApp {
    def main(args: Array[String]): Unit = {

        val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
        val tableEnvironment = StreamTableEnvironment.create(env)

        val stream = env.readTextFile("data/access.log").map(x => {
            val splits = x.split(",")
            Access(splits(0).trim.toLong, splits(1).trim, splits(2).trim.toDouble)
        })
        val table = tableEnvironment.fromDataStream(stream)
        tableEnvironment.createTemporaryView("access", table)

        val resultTable: Table = tableEnvironment.sqlQuery("select * from access where domain='ruozedata.com'")

        env.execute(getClass.getSimpleName)
    }
}
